package com.csw.spark

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo19Bro {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("acc")
      .setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")


    /**
      * 算子中的代码被封装成一个线程对象发送到Executor中执行
      * 当算子中使用算子外面的变量的时候，也需要将这个变量封装到线程对象中
      * ids集合就变成了一个变量副本(在算子这种只可以读，不可更改)
      *
      * 每一个task都需要封装一个变量副本，如果rdd的分区比较多，封装的变量副本就会很多
      * 就会导致线程对象变大，性能就会降低
      */

    //    val filterRDD: RDD[String] = studentRDD.filter(i => {
    //      val id: String = i.split(",")(0)
    //      ids.contains(id)
    //    })

    //    filterRDD.foreach(println)


    /**
      * 广播变量
      *
      */

    val ids = List("1500100010", "1500100016", "1500100005")

    //在Driver端将一个变量广播
    val broad: Broadcast[List[String]] = sc.broadcast(ids)

    val filterRDD: RDD[String] = studentRDD.filter(i => {
      val id: String = i.split(",")(0)
      //在算子内获取广播变量的数据
      val broadValue: List[String] = broad.value

      broadValue.contains(id)
    })
    filterRDD.foreach(println)

    /**
      *
      * * 算子中的代码被封装成一个线程对象发送到Executor端执行，广播变量的数据不会被封装到线程对象中
      *
      * 算子获取广播变量的流程：
      *     算子在执行的过程中如果使用到了广播变量，会向Executor获取，
      *     如果Executor有就直接使用，如果没有再去Driver端拉取广播变量的数据
      *
      * 使用广播变量和不使用广播变量的区别：
      * 1、不使用广播变量，每一个task都会封装一个变量的副本
      * 2、使用广播变量task中不会封装变量副本，每一个Executor中都会有变量副本
      *
      * 因为一般情况下  task的数量  >>  Executor的数量
      * 所以使用广播变量之后变量副本会减少
      */


    /**
      * 使用限制：
      *     广播变量只能在算子外部定义，只能在算子内部使用
      */
  }
}
